package com.sdyc.ndmp.cls.client;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.pattern.Patterns;
import akka.util.Timeout;
import com.sdyc.ndmp.cls.dtd.DataSentence;
import com.sdyc.ndmp.cls.dtd.KeywordCondition;
import com.sdyc.ndmp.cls.protocol.SentenceKeywordProtocol;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import java.io.IOException;
import java.io.Serializable;
import java.net.ServerSocket;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * <pre>
 * Created with IntelliJ IDEA.
 * User: zachary
 * Date: 2015/11/18
 * Time: 9:44
 * </pre>
 *
 * @author zachary
 */
public class CLSLAkkaClient implements SentenceKeywordProtocol {




    protected final String clientId = UUID.randomUUID().toString();



    private ActorRef actorRef = null;


    private ActorSystem actorSystem;



    protected int timeoutSeconds = 5;

    /**
     * Log
     */
    protected static Log LOG = LogFactory.getLog(CLSLAkkaClient.class);


    public CLSLAkkaClient() {
        this("localhost", 8080, 5);
    }


    public CLSLAkkaClient(String host, int port) {
        this(host, port, 5);
    }



    /**
     * @param host Server Host
     * @param port Server Akka Port
     * @param timeoutSeconds actor超时时间
     */
    public CLSLAkkaClient(String host, int port, int timeoutSeconds) {
        this.timeoutSeconds = timeoutSeconds;

        HashMap<String, Serializable> param = new HashMap<String, Serializable>();
        param.put("clientId", clientId);

        //Config load = ConfigFactory.load(conf);


        String remote = String.format("akka.tcp://CLSAkkaMaster@%s:%d", host, port);
        Properties prop = new Properties();
        prop.put("akka.loglevel", "INFO");
        prop.put("akka.actor.provider", "akka.remote.RemoteActorRefProvider");
        prop.put("akka.remote.enabled-transports", Arrays.asList("akka.remote.netty.tcp"));
        prop.put("akka.remote.transport", "akka.remote.netty.NettyRemoteTransport");
        prop.put("akka.remote.netty.tcp.hostname", "localhost");
        //prop.put("akka.remote.netty.tcp.port", "2553");
        prop.put("akka.actor.deployment.sentenceActor/*.remote", remote);


        int portFlag = 0;
        while (portFlag >= 0) {
            try {
                ServerSocket ss = new ServerSocket(Integer.valueOf("255" + portFlag));
                ss.close();
                prop.put("akka.remote.netty.tcp.port", "255" + portFlag);
                LOG.info("akka init port:" + "255" + portFlag);
                portFlag = -1;
            } catch (IOException ioe) {
                portFlag++;
            }
        }


        Config load = ConfigFactory.load(ConfigFactory.parseProperties(prop));

        actorSystem = ActorSystem.create("clsClient", load);
        actorRef = actorSystem.actorOf(Props.create(ClientActor.class, param), "sentenceActor");
        LOG.info("client init....");
    }




    /**
     * 此方法专门用来发送图片
     *
     * @param data
     * @return
     */
    @Override
    public Map<String, Object> send(DataSentence data) {
        Future<Object> future = Patterns.ask(actorRef,
                data,
                new Timeout(timeoutSeconds, TimeUnit.SECONDS));
        try {
            Map result = (Map) Await.result(future,
                    Duration.create(timeoutSeconds, TimeUnit.SECONDS));
            return result;
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }


    /**
     * 关闭服务
     */
    public void shutdown() {
        if (!actorSystem.isTerminated()) {
            actorSystem.shutdown();
            LOG.info("client is shutdown....");
        }
    }

    @Override
    public void setUp(KeywordCondition condition) {
        Future<Object> future = Patterns.ask(actorRef,
                condition,
                new Timeout(timeoutSeconds, TimeUnit.SECONDS));
        try {
            Object result = Await.result(future,
                    Duration.create(timeoutSeconds, TimeUnit.SECONDS));
            LOG.info("set up return " + result);
        } catch (Exception e) {
            //throw new IllegalStateException(e);
        }
    }

    @Override
    public void close() {
        shutdown();
    }



}
